package br.unifor.g2cl;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sf.jgcs.Annotation;
import net.sf.jgcs.ClosedSessionException;
import net.sf.jgcs.ExceptionListener;
import net.sf.jgcs.GroupConfiguration;
import net.sf.jgcs.JGCSException;
import net.sf.jgcs.Message;
import net.sf.jgcs.MessageListener;
import net.sf.jgcs.Service;
import net.sf.jgcs.ServiceListener;
import net.sf.jgcs.UnsupportedServiceException;
import net.sf.jgcs.membership.MembershipSession;


/**
 * DataSession with State transfer. It requires a TOTAL ORDER protocol in the
 * underlying gc toolkit. The algorithm is non-blocking (only the requester is
 * blocked while receiving state)
 * 
 * @author Henrique Teófilo
 * @author Leandro Sales
 */

/* TODO
 * Algumas situações deverão ser tratadas:
 * - Ninguém ter estado. (o erro que tava dando era "payload primeiro")
 * - O membro o qual eu pedi o estado (explicitamente) não ter o estado.
 * - o FIXME que eu botei lá na entrega das mensagens enfileiradas
 * - Ele necessita sempre receber mensagens próprias? se sim, avisar isso na doc.
 * - StateProvider cair
 * 
 * Consertar esses níveis de log que estão meio sem sentido.
*/
public class StateTransferDataSession implements IMarshalDataSession, ExceptionListener, MessageListener, ServiceListener {
	
	private static Logger logger = Logger.getLogger(StateTransferDataSession.class.getName());
	
	/** Header Format: BYTE */
	private static final byte MESSAGE = (byte) 1;
	
	/** Header Format: BYTE + STATE_ID. Avisar que quer o estado */
	private static final byte REQUEST_STATE = (byte) 2;
	
	/** Header Format: BYTE + STATE ID. Envia o estado */
	private static final byte SET_STATE = (byte) 3;
	
	/** Header Format: BYTE + STATE_ID. Pede o estado */ 
	private static final byte GET_STATE = (byte) 4;
	
	/** Header Format: BYTE + STATE_ID */
	private static final byte DESTROY_STATE = (byte) 5;

	/** Header Format: BYTE + STATE_ID. Enviado pelo candidato a state provider (pega o primeiro que receber) */
	private static final byte SET_STATE_PROVIDER = (byte) 6;
	
	
	private StateListener stateListener;
	private ExceptionListener realExceptionListener;
	private MessageListener realMessageListener;
	private ServiceListener realServiceListener;
	private IMarshalDataSession data;
	private MembershipSession control;
	/** service used to send state messages */
	private Service service;
	
	private List<G2CLMessage> queue;
	private boolean enqueue;
	private HashMap<String, byte[]> states;
	private String requestStateID;
	private SocketAddress stateProvider;


	
	

	/**
	 * 
	 * @param data
	 * @param service The Service must implement total order protocol
	 * @param control
	 */
	public StateTransferDataSession(IMarshalDataSession data, Service service, MembershipSession control) {
		this.service = service;
		this.data = data;
		this.control = control;
		this.states = new HashMap<String, byte[]>();
		this.queue = new LinkedList<G2CLMessage>();
		
		enqueue = false;
		requestStateID = null;
		stateProvider = null;
	}
	
	public void close() {
		data.close();
	}

	public Message createMessage() throws ClosedSessionException {
		return data.createMessage();
	}

	public GroupConfiguration getGroup() {
		return data.getGroup();
	}

	public void multicast(Message msg, Service service, Object cookie, Annotation... annotation) throws IOException,
																										UnsupportedServiceException {
		G2CLMessage message = (G2CLMessage) msg;
		message.addHeader(new byte[] { MESSAGE });
		data.multicast(message, service, cookie, annotation);
	}


	public void send(Message msg, Service service, Object cookie,
			SocketAddress destination, Annotation... annotation)
			throws IOException, UnsupportedServiceException {
		
		G2CLMessage message = (G2CLMessage) msg;
		message.addHeader(new byte[] { MESSAGE });
		data.send(message, service, cookie, destination, annotation);
	}
	
	
	/** Request the group state
	 * @param stateProvider Member to receive the state from. Null if you don't have any preference (you will get the state from anyone who has it)
	 * @throws IOException 
	 */
	public void requestState(SocketAddress stateProvider) throws IOException {
		if (stateListener == null) {
			throw new JGCSException("Must register the state listener");
		} else {
			this.requestStateID = control.getLocalAddress().toString() + "_" + System.currentTimeMillis();
			this.stateProvider = stateProvider;
			G2CLMessage m = createRequestStateMessage(this.requestStateID);
			this.queue.clear();
			enqueue = false;
			data.multicast(m, service, null, (Annotation[]) null);
		}
	}
	
	/**
	 * 
	 * @throws IOException
	 */
	public void requestState() throws IOException {
		SocketAddress addr;
		try {
			int coordRank = control.getMembership().getCoordinatorRank();
			addr = control.getMembership().getMemberAddress(coordRank);
			
			logger.info("Getting state from:" + addr);
			
		} catch (UnsupportedOperationException e) {
			logger.info("Cannot get information about group coordinator. Finding out a member to request state...");
			addr = null;
		}
		requestState(addr);
	}

	public Object onMessage(Message msg) {
				
		G2CLMessage message = (G2CLMessage) msg;

		byte[] header = message.removeHeader();
		
		switch (header[0]) {
		case MESSAGE:
			if (header[0] == MESSAGE) {				
				if (enqueue) {
					queue.add(message);
				} else {
					return realMessageListener.onMessage(msg);
				}
			}
			break;

		case SET_STATE_PROVIDER:
			handle_SetStateProviderMessage(message);
			break;

		case REQUEST_STATE:
			handle_RequestStateMessage(message);
			break;

		case GET_STATE:
			handle_GetStateMessage(message);
			break;

		case SET_STATE:
			handle_SetStateMessage(message);
			break;

		case DESTROY_STATE:
			handle_DestroyStateMessage(message);
			break;
		}

		return null;
	}

	
	/* ********************************************************************** *
	 * ***************** Methods to handle incoming messages **************** *
	 * ********************************************************************** */
	
	private void handle_SetStateProviderMessage(G2CLMessage msg) {
		if (requestStateID != null && stateProvider == null) {

			String sId = byteToString(msg.getPayload());

			logger.log(Level.FINER, "SET_STATE_PROVIDER ID: '" + sId + "' received from: " + msg.getSenderAddress());

			if (sId.equals(requestStateID)) {
				this.stateProvider = msg.getSenderAddress();
				askForState();
			}
		}
	}
	
	private void handle_RequestStateMessage(G2CLMessage msg) {
		if (msg.getSenderAddress().equals(control.getLocalAddress())) {
			// do not notify the message listener
			if (logger.isLoggable(Level.FINER)) {
				logger.log(Level.FINER, "Enqueuing new messages");
			}
			enqueue = true; // enqueue next messages
			if (stateProvider != null) {
				askForState();
			}
		} else {
			String id = byteToString(msg.getPayload());
			// retrieving state from the application
			
			if (stateListener.hasState()) {
				byte[] myStateCopy = stateListener.stateOutgoing();
				logger.log(Level.FINER,"Copying and saving my state. ID: '" + id + "'");
				states.put(id, myStateCopy);
				
				try {
					G2CLMessage m = createSetStateProviderMessage(id);
					logger.log(Level.FINER, "Sending SET_STATE_PROVIDER message");
					data.multicast(m, service, null, (Annotation[]) null);
				} catch (IOException e) {
					logger.log(Level.FINER, "Error sending SET_STATE_PROVIDER message", e);
				}
			} else {
				logger.info("I do not have the state.");
			}
		}
	}
	
	private void handle_GetStateMessage(G2CLMessage msg) {
		try {
			logger.log(Level.FINER, "Received GET_STATE from " + msg.getSenderAddress());
			String id = byteToString(msg.getPayload());
			G2CLMessage m = createSetStateMessage(id);
			logger.log(Level.FINER, "Sending my copy of the state");
			data.send(m, service, null, msg.getSenderAddress(), (Annotation[]) null);
		} catch (IOException e) {
			logger.log(Level.SEVERE, "error sending my state", e);
		}
	}
	
	private void handle_SetStateMessage(G2CLMessage msg) {
		String rId = byteToString(msg.removeHeader());

		if (rId.equals(requestStateID)) {
			try {
				byte[] state = msg.getPayload();
				logger.log(Level.FINER, "State retrieved from " + msg.getSenderAddress());
				stateListener.stateIncoming(state);
				G2CLMessage m = createDestroyStateMessage();
				data.multicast(m, service, null, (Annotation[]) null);

			} catch (Exception e) {
				logger.log(Level.SEVERE, "error retrieving state", e);
			}
		} else {
			logger.log(Level.FINER, "The state ID received '" + rId + "' is not" +
			" equal to the request state id. Discarding...");
		}
	}

	
	private void handle_DestroyStateMessage(G2CLMessage msg) {
		if (enqueue) {
			deliverQueuedMessages();
			enqueue = false;
			requestStateID = null;
			stateProvider = null;
			queue.clear();
		} else {
			String id = byteToString(msg.getPayload());
			logger.log(Level.FINER, "Destroying state: '" + id + "'");
			states.remove(id);
		}
	}
	
	/* ********************************************************************** *
	 * ********************** Auxiliary methods ***************************** *
	 * ********************************************************************** */
	
	
	/** Ask state provider for the group state */
	private void askForState() {
		try {
			G2CLMessage m = createGetStateMessage();
			logger.log(Level.FINER,"asking " + this.stateProvider + " for the group state");
			data.send(m, service, null, this.stateProvider, (Annotation[]) null);
		} catch (Exception e) {
			logger.log(Level.SEVERE,"error requesting state", e);
		}
	}


	/** Deliver all queued messages */
	private void deliverQueuedMessages() {
		logger.log(Level.FINER, "Delivering queued messages");
		// FIXME It must be synchronized with the enqueuing of new messages
		for (G2CLMessage m: queue) {
			realMessageListener.onMessage(m);
		}
	}
	

	private G2CLMessage createSetStateProviderMessage(String strId) throws ClosedSessionException {
		byte[] id = stringToByte(strId);
		G2CLMessage m = (G2CLMessage)createMessage();
		m.setSenderAddress(control.getLocalAddress());
		m.setPayload(id);
		m.addHeader(new byte[] { SET_STATE_PROVIDER });
		return m;
	}
	
	private G2CLMessage createRequestStateMessage(String requestStateID) throws ClosedSessionException {
		logger.log(Level.FINER,"Creating request state message with ID: '" + requestStateID + "'");
		byte[] id = stringToByte(requestStateID);
		G2CLMessage m = (G2CLMessage)createMessage();
		m.setSenderAddress(control.getLocalAddress());
		m.setPayload(id);
		m.addHeader(new byte[] { REQUEST_STATE });
		return m;
	}
	
	private G2CLMessage createDestroyStateMessage() throws ClosedSessionException {
		byte[] id = stringToByte(this.requestStateID);
		G2CLMessage m = (G2CLMessage)createMessage();
		m.setSenderAddress(control.getLocalAddress());
		m.setPayload(id);
		m.addHeader(new byte[] { DESTROY_STATE });
		return m;
	}

	private G2CLMessage createSetStateMessage(String id) throws ClosedSessionException {
		G2CLMessage m = (G2CLMessage) createMessage();
		m.setSenderAddress(control.getLocalAddress());
		m.setPayload(states.get(id));
		m.addHeader(stringToByte(id));
		m.addHeader(new byte[] { SET_STATE });
		return m;
	}
	
	private G2CLMessage createGetStateMessage() throws ClosedSessionException {
		byte[] id = stringToByte(this.requestStateID);
		G2CLMessage m = (G2CLMessage) createMessage();
		m.setSenderAddress(control.getLocalAddress());
		m.setPayload(id);
		m.addHeader(new byte[] { GET_STATE });
		return m;
	}
	
	
	private static String byteToString(byte[] barr) {
		try {
			return new String(barr, "ascii");
		} catch (UnsupportedEncodingException e) {
			logger.log(Level.SEVERE,"error converting byte array into string", e);
		}
		return null;
	}
	
	private static byte[] stringToByte(String str) {
		try {
			return str.getBytes("ascii");
		} catch (UnsupportedEncodingException e) {
			logger.log(Level.SEVERE,"error converting string into byte array", e);
		}
		return null;
	}
	
	
	/* ********************************************************************** *
	 * **************************** LISTENERS ******************************* *
	 * ********************************************************************** */
	
	public void setExceptionListener(ExceptionListener exception)
	throws ClosedSessionException {
		this.realExceptionListener = exception;
		data.setExceptionListener(this);
	}

	public void setMessageListener(MessageListener listener) throws ClosedSessionException {
		this.realMessageListener = listener;
		data.setMessageListener(this);
	}

	public void setServiceListener(ServiceListener listener)
	throws ClosedSessionException {
		this.realServiceListener = listener;
		data.setServiceListener(this);
	}

	public void onException(JGCSException exception) {
		realExceptionListener.onException(exception);
	}
	public void onServiceEnsured(Object context, Service service) {
		realServiceListener.onServiceEnsured(context, service);
	}
	
	public void setStateListener(StateListener stateListener) {
		this.stateListener = stateListener;
	}	
}